package org.lzfto.flink.demo.business.flinkTest.test;

import java.util.concurrent.ExecutionException;

import org.apache.flink.api.common.JobStatus;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.lzfto.flink.demo.business.flinkTest.entity.TestInfo;
import org.lzfto.flink.demo.common.result.PubResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*
*@author lzfto
*@创建时间 2025年7月9日 14:28:33
*/
public class BaseTestImpl implements FlinkTestInterface {
	
	private final  Logger LOGGER = LoggerFactory.getLogger(getClass());
	
	protected StreamExecutionEnvironment env;
	protected JobClient jobClient;
	

	@Override
	public TestInfo getTestInfo() {
		return null;
	}

	@Override
	public Object run() throws Exception {
		return null;
	}
	
	@Override
	public PubResult<Boolean> clean() {
		if (this.jobClient != null) {
			JobStatus jobStatus = null;
			try {
				jobStatus = this.jobClient.getJobStatus().get();
			} catch (InterruptedException | ExecutionException e) {
				e.printStackTrace();
				return PubResult.ok("停止作业失败,无法获取作业状态:" + e.getMessage(),false);
			}
			catch(Exception e) {
				String errMsg=e.getMessage();
				//注意：这个处理方式目前只在1.20的本地运行上测试过。
				if (errMsg.equals("MiniCluster is not yet running or has already been shut down.")) {
					return PubResult.ok("微集群已经停止",true);
				}
				else {
					return PubResult.ok("停止作业失败,无法获取作业状态:" + errMsg,false);
				}
			}
			try {
				if (!(jobStatus.equals(JobStatus.CANCELED) || jobStatus.equals(JobStatus.FINISHED))) {
					String jobId = jobClient.getJobID().toString();
					jobClient.cancel();
					if (this.env != null) {
						this.env.close();
					}
					LOGGER.warn("停止作业(id={})及其相关环境!", jobId);
					jobClient = null;
					env = null;
					return PubResult.ok("已经成功停止作业(jobId="+jobId+")",true);
				} else {
					return PubResult.ok("作业已经停止了!无需再次停止",true);
				}
			} catch (Exception e) {
				LOGGER.warn("停止作业失败");
				e.printStackTrace();
				return PubResult.ok("停止作业失败:" + e.getMessage(),true);
			}
		} else {
			return PubResult.ok("停止作业失败!作业尚未创建",true);
		}
	}
	
	@Override
	public Boolean isRunning() {
		if (this.jobClient != null) {
			JobStatus status = null;
			try {
				status = this.jobClient.getJobStatus().get();
				if (status.equals(JobStatus.RUNNING)) {
					return true;
				}
			} catch (InterruptedException e) {
				LOGGER.error("获取作业运行状态失败，原因:{}", e.getMessage());
			} catch (ExecutionException e) {
				LOGGER.error("获取作业运行状态失败，原因:{}", e.getMessage());
			}
		}
		return false;
	}

}
